<?php
/**
 * Created by PhpStorm.
 * User: wyq
 * Date: 2022/2/16
 * Time: 16:34
 */
namespace MyRabbitMq;

require_once "./vendor/autoload.php";

use \PhpAmqpLib\Connection\AMQPStreamConnection;
use \PhpAmqpLib\Message\AMQPMessage;

class MyRabbitMq
{
    protected $connection; //连接
    protected $channel; //信道

    public function __construct($user ="wyqgg", $password ="123456",$port = 5672,  $vHost = '/' )
    {

//        $host = PHP_OS == "Linux" ? '127.0.0.1' : '10.0.0.200';
        $host = '127.0.0.1';
        //初始化连接
        $this->connection = new AMQPStreamConnection($host, $port, $user, $password, $vHost);
        //初始化信道
        $this->channel = $this->connection->channel();
    }

    //获取信道
    public function getChannel()
    {
        return $this->channel;
    }

    //创建交换机
    public function setExchange($exchange_name, $type, $passive = false, $durable = false, $auto_delete = false, $internal = false, $nowait = false, $arguments = array(), $ticket = null)
    {
        $this->channel->exchange_declare($exchange_name, $type, $passive, $durable, $auto_delete, $internal, $nowait, $arguments, $ticket);
    }

    //创建队列
    public function setQueue($queue, $passive, $durable = false, $exclusive = false, $auto_delete = false, $nowait = false, $arguments = array(), $ticket = null)
    {
        $this->channel->queue_declare($queue, $passive, $durable, $exclusive, $auto_delete, $nowait, $arguments, $ticket);
    }

    //绑定交换机和队列
    public function bindQueueExchange($queue_name, $exchange_name, $router_key = '', $nowait = false, $arguments = array(), $ticket = null)
    {
        $this->channel->queue_bind($queue_name, $exchange_name, $router_key, $nowait, $arguments, $ticket);
    }

    //发送消息
    public function sendMsg($msg, $properties = [], $exchange = '', $routing_key = '', $mandatory = false, $immediate = false, $ticket = null)
    {
        $message = new AMQPMessage($msg, $properties);
        $this->channel->basic_publish($message, $exchange, $routing_key, $mandatory, $immediate, $ticket);
    }

    //消费消息
    public function consume($queue = '', $callback = null, $consumer_tag = '', $no_local = false, $no_ack = false, $exclusive = false, $nowait = false, $ticket = null, $arguments = array())
    {
        $this->channel->basic_consume($queue, $consumer_tag, $no_local, $no_ack, $exclusive, $nowait, $callback, $ticket, $arguments);
        while ($this->channel->is_consuming()) {
            $this->channel->wait();
        }
    }
    //设置confirm
    public function setConfirm()
    {
        $this->channel->confirm_select();
    }

    //发送成功
    public function setAckHandler($callback)
    {
        $this->channel->set_ack_handler($callback);
    }

    //发送失败
    public function setNAckHandler($callback)
    {
        $this->channel->set_nack_handler($callback);
    }

    public function waitForPendingAckReturns($time)
    {
        $this->channel->wait_for_pending_acks_returns($time);
    }
    //断开连接
    public function close()
    {
        $this->channel->close();
        $this->connection->close();
    }

}